package com.google.code.chaotic.storedprocs;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.Arrays;
import java.util.HashSet;
import java.util.LinkedList;

import org.postgresql.pljava.ResultSetProvider;
import org.postgresql.pljava.jdbc.SQLUtils;

public class CommutingGroupsHSProvider implements ResultSetProvider {

	// TODO: Most of the literal values must be defined as parameters!

	private final static String[] ALGO_QUERIES = new String[] {

			"SELECT 		 h1.id AS id, "
					+ "      array_accum (h2.id) AS mates_ids, "
					+ " 	 1 AS density, "
					+ " 	 count(h2.id) AS mate_count "
					+ "FROM tmp_habit AS h1, " //
					+ "     tmp_habit AS h2  " //
					+ "WHERE h1.id <> h2.id "
					+ " 	 AND ST_DWithin(h1.orig_gis, h2.orig_gis, 2000) "
					+ "      AND ST_DWithin(h1.dest_gis, h2.dest_gis, 2000) "
					+ "GROUP BY h1.id, h1.orig_gis "
					+ "ORDER BY mate_count DESC ",

			"SELECT 		  h1.id AS id, "
					+ "       array_accum (h2.id) AS mates_ids, "
					+ "       sum( 1 / CASE ST_Distance(h1.orig_gis, h2.orig_gis) 		"
					+ "		   				WHEN 0 THEN 1 								"
					+ "		   				ELSE ST_Distance(h1.orig_gis, h2.orig_gis) 	"
					+ "		   		   END ) AS density "
					+ "FROM tmp_habit AS h1, "
					+ "     tmp_habit AS h2  " //
					+ "WHERE h1.id <> h2.id   "
					+ "      AND ST_DWithin(h1.orig_gis, h2.orig_gis, 1500)  	"
					+ "      AND ST_DWithin(h1.dest_gis, h2.dest_gis, 2000)  	"
					+ "      AND ST_Distance(h2.orig_gis, h2.dest_gis) > 3500 	"
					+ "GROUP BY h1.id, h1.orig_gis 	" //
					+ "ORDER BY density DESC " };

	public static ResultSetProvider list(int algoNo) throws SQLException {
		try {
			Class.forName(ProcUtils.class.getName());
			return new CommutingGroupsHSProvider(algoNo);

		} catch (Throwable e) {
			throw ProcUtils.wrapAndLog(e);
		}
	}

	private final int algoNo;
	private Statement stmt1;
	private Statement stmt2;

	private final HashSet<String> idSet = new HashSet<String>(1000);
	private LinkedList<Row> processedRows = new LinkedList<Row>();
	private boolean lastFetch;

	private class Row {
		String id;
		double density;
		String[] candidateIds;
	}

	// private int lastRow = 0;

	public CommutingGroupsHSProvider(int algoNo) throws SQLException {
		this.algoNo = algoNo;
		Connection conn = SQLUtils.getDefaultConnection();
		stmt1 = conn.createStatement();
		
		String //
		sqlStmt1 = "CREATE TEMP TABLE tmp_habit"
				+ " (id VARCHAR, orig_gis GEOMETRY, dest_gis GEOMETRY)";
		String //
		sqlStmt2 = "CREATE INDEX tmp_habit_orig_index "
				+ " ON tmp_habit USING gist (orig_gis)";
		String //
		sqlStmt3 = "CREATE INDEX tmp_habit_dest_index "
				+ " ON tmp_habit USING gist (dest_gis)";
		String //
		sqlStmt4 = "INSERT INTO tmp_habit SELECT id, orig_gis, dest_gis"
				+ " FROM commuting_habit";
		
		stmt1.execute(sqlStmt1);
		stmt1.execute(sqlStmt2);
		stmt1.execute(sqlStmt3);
		stmt1.execute(sqlStmt4);
	}

	public boolean assignRowValues(ResultSet receiver, int currentRow)
			throws SQLException {

		Statement stmt3 = null;
		try {
			Connection conn = SQLUtils.getDefaultConnection();
			if (processedRows.size() == 0) {
				stmt2 = conn.createStatement();
				int fetchSize = 1000;

				StringBuilder sql = new StringBuilder();
				sql.append(ALGO_QUERIES[algoNo - 1]);
				sql.append(" LIMIT ");
				sql.append(fetchSize);

				ResultSet rs = stmt2.executeQuery(sql.toString());
				LinkedList<Row> rows = new LinkedList<Row>();
				while (rs.next())
					rows.add(extractOneRow(rs));

				lastFetch = rows.size() < fetchSize;
				process(rows);
				SQLUtils.close(stmt3);
			}
			if (processedRows.size() == 0)
				if (lastFetch)
					return false;
				else
					return assignRowValues(receiver, currentRow);

			Row row = processedRows.poll();

			// String[] matesIds = new String[Math
			// .min(10, row.candidateIds.length)];
			// System.arraycopy( //
			// row.candidateIds, 0, matesIds, 0, matesIds.length);
			String[] matesIds = row.candidateIds;

			StringBuilder matesStr = new StringBuilder(//
					(matesIds.length + 2) * row.id.length());
			matesStr.append("ARRAY[");

			StringBuilder matesCsv = new StringBuilder();
			for (String mateId : matesIds) {
				ProcUtils.appendToArrayStr(matesCsv, mateId);
				matesCsv.append(',');
			}
			// if (matesCsv.length() > 0)
			// matesCsv.deleteCharAt(matesCsv.length() - 1);
			ProcUtils.appendToArrayStr(matesCsv, row.id);

			matesStr.append(matesCsv).append("]");
			String //
			sqlStmt3 = "DELETE FROM tmp_habit " //
					+ "  	WHERE id = ANY (" + matesStr + ")";
			stmt3 = conn.createStatement();
			stmt3.execute(sqlStmt3);
			// String //
			// sqlStmt3 = "DELETE FROM tmp_habit " //
			// + "  	WHERE id = ANY (?)";
			// stmt3 = conn.prepareStatement(sqlStmt3);
			// ((PreparedStatement)stmt3).setObject(1, arrayObject);
			// stmt3.execute(sqlStmt3);

			receiver.updateString(1, row.id);
			receiver.updateInt(2, matesIds.length);
			// receiver.updateObject(3, mates);
			receiver.updateObject(3, Arrays.deepToString(matesIds));
			return true;

		} catch (Throwable e) {
			throw ProcUtils.wrapAndLog(e);

		} finally {
			SQLUtils.close(stmt2);
		}
	}

	private void process(LinkedList<Row> rows) {
		if (rows.size() == 0)
			return;
		processedRows.clear();
		Row row = rows.poll();

		while (row != null) {
			boolean goodGroup = processOneRow(row);
			// if (goodGroup) {
			// int skipCount = (int) Math.min(//
			// rows.size(), row.candidateIds.length * 0.75);
			// rows.subList(0, skipCount).clear();
			// }
			row = rows.poll();
		}
	}

	private boolean processOneRow(Row row) {
		if (idSet.contains(row.id)) {
			lastFetch = lastFetch && row.candidateIds.length == 0;
			return false;
		}
		HashSet<String> matesIdSet = new HashSet<String>(10);
		for (String id : row.candidateIds) {
			if (!idSet.contains(id))
				matesIdSet.add(id);

			if (matesIdSet.size() == 10)
				break;
		}
		if (!lastFetch && matesIdSet.size() < 6 && idSet.size() != 0)
			return false;
		row.candidateIds = new String[matesIdSet.size()];
		matesIdSet.toArray(row.candidateIds);
		idSet.add(row.id);
		idSet.addAll(matesIdSet);
		processedRows.add(row);
		return true;
	}

	private Row extractOneRow(ResultSet rs) throws SQLException {
		Row row = new Row();
		row.id = rs.getString(1);
		Object arrayObject = rs.getObject(2);
		row.candidateIds = (String[]) arrayObject;
		row.density = rs.getDouble(3);
		return row;
	}

	public void close() throws SQLException {
		if (stmt1 != null) {
			stmt1.execute("DROP TABLE tmp_habit");
			stmt1.close();
		}
	}

}
